package kassiopeia.agent.seda;

import kassiopeia.agent.internal.metrics.MetricsAggregator;
import kassiopeia.agent.seda.config.ConfiguredTask;
import kassiopeia.agent.seda.config.SedaConnection;
import kassiopeia.agent.seda.config.State;
import kassiopeia.agent.seda.config.TaskConfigurator;
import kassiopeia.agent.seda.threads.SedaPool;
import kassiopeia.commons.exception.SedaException;
import kassiopeia.commons.model.task.TaskDescriptor;
import kassiopeia.plugin.Event;
import org.apache.log4j.Logger;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import javax.annotation.PreDestroy;
import java.util.List;

@Component("seda")
public class SEDA {

    private final Logger log = Logger.getLogger(SEDA.class);

    private State state = State.STOPPED;
    private ConfiguredTask configuredTask;
    private SedaPool sedaPool;

    @Autowired
    private TaskConfigurator taskConfigurator;

    public synchronized void start(TaskDescriptor task) {
        log.info("Starting SEDA execution environment...");
        if (task == null) {
            throw new SedaException("Task is null");
        }
        if (state != State.STOPPED) {
            throw new SedaException("Cannot start SEDA (SEDA is in " + state + " state)");
        }
        configuredTask = taskConfigurator.configure(task);
        sedaPool = new SedaPool(configuredTask.allStages());
        sedaPool.start();
        state = State.RUNNING;
    }

    public synchronized void stop() {
        log.info("Stopping execution environment...");
        if (state != State.RUNNING) {
            throw new SedaException("SEDA is not running");
        }
        state = State.TERMINATING;
        sedaPool.shutdown();
        state = State.STOPPED;
        log.info("Execution environment stopped");
    }

    @PreDestroy
    public synchronized void close() {
        if (state == State.RUNNING) {
            stop();
        }
    }

    public void send(String instanceId, String outputName, Event event) {
        List<SedaConnection> connections = configuredTask.getConnectionsFor(instanceId, outputName);
        if (connections == null || connections.isEmpty()) {
            log.warn("Trying to send event to non existing queue (from: " + instanceId + "[" + outputName + "])");
        } else {
            for (SedaConnection conn : connections) {
                conn.send(event);
                if (MetricsAggregator.isEnabled()) {
                    MetricsAggregator.countOutgoingEvent(conn);
                }
            }
        }
    }

    public State getState() {
        return state;
    }
}
